#pragma once
#include <mqcomm/Log.hpp>
#include <mqcomm/Helper.hpp>
#include <mqcomm/Message.pb.h>
#include <iostream>
#include <string>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <cassert>
#include <list>
using namespace std;


namespace rbmq {
    #define DATAFILE_SUFFIX ".mqd"
    #define TEMPFILE_SUFFIX ".mqd.temp"
    typedef shared_ptr<mqNs::Message> MessagePtr;
    // 消息持久化存储管理类
    class MessageMapper {
    public :
        MessageMapper(string& basedir, const string& qname) 
            :_qname(qname)
        {
            // 判断basedir后面有没有'/'， 如果没有先加上
            if (basedir.back() != '/') {
                basedir += "/";
            }
            // 创建目录
            assert(FileHelper::createDirectory(basedir));
            _datafile = basedir + qname + DATAFILE_SUFFIX;
            _tempfile = basedir + qname + TEMPFILE_SUFFIX;
            // 创建消息存储文件
            assert(createMessageFile());
        }

        // 消息文件的创建
        bool createMessageFile() {
            if (!FileHelper::createFile(_datafile)) {
                LOG(Fatal) << "创建消息存储文件失败!" << endl;
                return false;
            }
            return true;
        }
        // 消息文件的杀出
        bool removeMessageFile() {
            // 将_datafile和_tempfile都删除了
            if (!FileHelper::removeFile(_datafile)) {
                LOG(Debug) << "删除消息存储文件失败！" << endl;
                return false;
            }
            if (!FileHelper::removeFile(_tempfile)) {
                LOG(Debug) << "删除临时文件失败！" << endl;
                return false;
            }
            return true;
        }

        // 新增消息
        bool insert(MessagePtr& msg) {
            return insert(_datafile, msg);
        }
        // 消息的删除
        bool remove(MessagePtr& msg) {
            // 删除消息其实并不是真正删除，而是将修改后的msg覆盖原文件中的位置
            // 1、将msg中的有效标志位修改成"0"
            // string payload_body = msg->payload().SerializeAsString();
    
            msg->mutable_payload()->set_valid("0");
            // 2、对message进行序列化
            string payload_body = msg->payload().SerializeAsString();
            // 3、将序列化后的消息，覆盖的写入到文件中原来的位置(写入之前先判断，序列化的结果和msg原本的length相不相同，如果不相同也是不能写入的)
            if (payload_body.size() != msg->length()) {
                LOG(Debug) << "不能删除该消息, 因为该消息序列化后与原消息长度不一致！" << endl;
                return false;
            }
            if (!FileHelper::write(_datafile, &payload_body[0], msg->offset(), payload_body.size())) {
                LOG(Error) << "删除消息是, 重新写入消息存储文件失败！" << endl;
                return false;
            }
            return true;
        }
        // 垃圾回收机制
        list<MessagePtr> gc() {
            // 1、加载出文件中所有的有效数据
            list<MessagePtr> result;
            if (!FileHelper::createFile(_tempfile)) {
                LOG(Error) << "创建临时文件失败！" << endl;
                return result;
            }
            size_t offset = 0, msg_size;
            size_t filesize = FileHelper::size(_datafile);
            while (offset < filesize) {
                // 每次先读出消息长度
                if (!FileHelper::read(_datafile, (char*)&msg_size, offset, sizeof(size_t))) {
                    LOG(Debug) << "读取消息长度失败！" << endl;
                    return result;
                }
                offset += sizeof(size_t);
                // 读取消息
                string payload_body(msg_size, '\0');
                if (!FileHelper::read(_datafile, &payload_body[0], offset, msg_size)) {
                    LOG(Debug) << "读取消息数据出错!" << endl;
                    return result;
                }
                offset += msg_size;
                // 对消息进行反序列化
                MessagePtr new_msg = make_shared<mqNs::Message>();
                if (!new_msg->mutable_payload()->ParseFromString(payload_body)) {
                    LOG(Debug) << "对消息进行反序列化失败！" << endl;
                    return result;
                }
                // 判断是否是有效消息，如果不是有效消息，则不需要存储到临时文件中
                if (new_msg->payload().valid() == "0") {
                    continue;
                }
                // 2、将有效数据序列化后，存储到一个临时文件中(同时还要修改消息的属性)
                if (!insert(_tempfile, new_msg)) {
                    return result;
                }
                result.push_back(new_msg);
            }

            // 3、删除原文件
            if (!FileHelper::removeFile(_datafile)) {
                LOG(Debug) << "删除原文件失败！" << endl;
                return result;
            }
            // 4、重命名临时文件名为原文件名
            if (!FileHelper::rename(_tempfile, _datafile)) {
                LOG(Debug) << "重命名临时文件失败！" << endl;
                return result;
            }
            // 5、返回新的有效数据
            return result;
        }
        // // 加载所有的有效数据
        // bool load(list<MessagePtr>& result) {

        // }
    private :
        bool insert(const string& filename, MessagePtr& msg) {
            // 新增数据都是添加在文件末尾的
            // 所以需要先获取文件的大小
            // 1、获取文件大小
            size_t filesize = FileHelper::size(filename);
            // 2、对消息进行序列化(真正存储的是msg中的payload，所以需要序列化的是payload)
            string payload_body = msg->payload().SerializeAsString();
            // 3、将数据写入到文件的指定位置

            // 先写入消息长度
            size_t msg_size = payload_body.size();
            if (!FileHelper::write(filename, (char*)&msg_size, filesize, sizeof(size_t))) {
                LOG(Error) << "将消息长度写入文件: " << filename << " 失败!" << endl;
                return false;
            }

            // 再写入消息
            if (!FileHelper::write(filename, &payload_body[0], filesize + sizeof(size_t), payload_body.size())) {
                LOG(Error) << "将消息写入文件: " << filename << " 失败！" << endl;
                return false;
            }
            // 4、更新msg中的实际存储信息
            msg->set_offset(filesize + sizeof(size_t));
            msg->set_length(payload_body.size());
            return true;
        }
    private :
        string _qname; // 队列名称
        string _datafile; // 消息存储文件名
        string _tempfile; // 消息临时存储文件名
    };
    // ----------------------------------------------------------------------------------------------------------------------------------------


    // 队列消息管理类
    class QueueMessage {
    public :
        typedef shared_ptr<QueueMessage> ptr;
        QueueMessage(string& basedir, const string& qname) 
            :_qname(qname)
            ,_mapper(basedir, qname)
            ,_valid_count(0)
            ,_total_count(0)
        {

        }

        // 新增一条消息
        bool insert(const mqNs::BasicProperties* bp, const string& body, bool queue_is_durable) {
            unique_lock<mutex> lock(_mutex);
            // 1、构造消息对象
            MessagePtr msg = make_shared<mqNs::Message>();

            msg->mutable_payload()->set_body(body);

            // 设置属性前，先判断bp是否为空，不为空就是用bp的属性构造对象，为空就是用delivery_mode中的属性构造对象
            if (bp) {
                mqNs::DeliveryMode mode;
                if (bp->delivery_mode() == mqNs::DeliveryMode::DURABLE && queue_is_durable) {
                    mode = mqNs::DeliveryMode::DURABLE;
                } else {
                    mode = mqNs::DeliveryMode::UNDURABLE;
                }
                if (msg->mutable_payload()->mutable_properties() == nullptr) {
                }

                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            } else {

                mqNs::DeliveryMode mode;

                if (queue_is_durable) {

                    mode = mqNs::DeliveryMode::DURABLE;
                } else {
                    mode = mqNs::DeliveryMode::UNDURABLE;
                }

                msg->mutable_payload()->mutable_properties()->set_id(UuidHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            string msg_id = msg->mutable_payload()->mutable_properties()->id();
            // 2、判断消息是否需要持久化存储

            if (msg->payload().properties().delivery_mode() == mqNs::DeliveryMode::DURABLE) {
                // 3、如果需要，进行持久化存储
                // 先将有效标志位置为“1”
                msg->mutable_payload()->set_valid("1");

                if (!_mapper.insert(msg)) {
                    LOG(Debug) << "消息: " << msg_id << " 进行持久化存储失败!" << endl;
                    return false;
                }

                _durable_msgs.insert(make_pair(msg_id, msg));
                _total_count++;
                _valid_count++;
            }
            // 4、内存的管理

            _send_msgs.push_back(msg);

            return true;
        }
        // 确认一条消息(即从待确认队列中删除指定id的消息)
        bool remove(const string& msg_id) {
            unique_lock<mutex> lock(_mutex);
            // 每次删除一条消息之后，判断是否需要垃圾回收

            // 1、从待确认消息哈希表中，查找这条消息，如果没有，就不用删除了
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end()) {
                return true;
            }
            // 2、根据消息的持久化模式，确定是否需要删除持久化数据
            if (it->second->payload().properties().delivery_mode() == mqNs::DeliveryMode::DURABLE) {
                // 3、如果有需要，删除持久化信息，并且删除后判断是否需要垃圾回收
                if (!_mapper.remove(it->second)) {
                    LOG(Debug) << "删除持久化消息: " << msg_id << " 失败!" << endl;
                    return false;
                }
                _durable_msgs.erase(msg_id);
                _valid_count--;
                gc();
            }

            // 4、删除内存中的信息
            _waitack_msgs.erase(msg_id);
            return true;
        }

        // 获取一条队首消息
        MessagePtr front() {
            unique_lock<mutex> lock(_mutex);
            // 获取一条队首消息，即从_send_msgs中取出队首消息
            MessagePtr msg = _send_msgs.front();
            _send_msgs.pop_front();
            // 将该消息，推送到待确认消息哈希表中，等到确认后再删除
            _waitack_msgs.insert(make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        // 获取可推送消息数量
        size_t push_count() {
            unique_lock<mutex> lock(_mutex);
            return _send_msgs.size();

        }

        // 获取持久化存储的消息数量
        size_t durable_count() {
            unique_lock<mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        
        // 获取总体的消息数量
        size_t total_count() {
            unique_lock<mutex> lock(_mutex);
            return _total_count;

        }
        // 获取待确认的消息数量
        size_t waitack_count() {
            unique_lock<mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        // 清理所有的数据
        void clear() {
            unique_lock<mutex> lock(_mutex);
            _mapper.removeMessageFile();
            _send_msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _total_count = 0;
            _valid_count = 0;
        }

        // 恢复历史消息
        void recovery() {
            unique_lock<mutex> lock(_mutex);
            _send_msgs = _mapper.gc();
            for (auto& msg : _send_msgs) {
                _durable_msgs.insert(make_pair(msg->payload().properties().id(), msg));
            }
            _total_count = _valid_count = _send_msgs.size();
        }
    private :
        bool gc_check() {
            return (_total_count > 2000 && _valid_count * 10 / _total_count < 5);
        }
        // 垃圾回收机制
        void gc() {
            if (!gc_check()) {
                return;
            }
            // 1、进行垃圾回收，新的有效的消息链表
            list<MessagePtr> msgs = _mapper.gc();
            // 返回链表之后，还需要判断链表中的消息的合理性，因为返回的心得有效消息一定是要在_durable_msgs中存在的，
            for (auto& msg : msgs) {
                string msg_id = msg->payload().properties().id();
                auto it = _durable_msgs.find(msg_id);
                if (it == _durable_msgs.end()) {
                    LOG(Debug) << "垃圾回收后, 发现一条消息: " << msg_id << " 没有在内存中进行管理" << endl;
                    // 如果不存在则将该消息重新加入到_qurable_msg和_send_msgs中
                    _durable_msgs.insert(make_pair(msg_id, msg));
                    _send_msgs.push_back(msg);
                    continue;
                }
                // 找着了就更新这条消息的实际存储
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }

            // 3、更新当前的有效消息数量以及总的消息数量
            _total_count = _valid_count = msgs.size();
        }

    private :
        mutex _mutex; // 互斥锁
        string _qname;
        size_t _valid_count; // 有效持久化的消息数量
        size_t _total_count; // 总体持久化的消息数量
        MessageMapper _mapper; // 队列消息持久化管理句柄
        list<MessagePtr> _send_msgs; // 待推送消息链表
        unordered_map<string, MessagePtr> _durable_msgs; // 持久化消息的消息id与消息指针的哈希
        unordered_map<string, MessagePtr> _waitack_msgs; // 待确认消息id与消息指针的哈希
    };
    // ----------------------------------------------------------------------------------------------------------------------------------------
    // 对外提供的总体消息管理类
    class MessageManager {
    public :
        typedef shared_ptr<MessageManager> ptr;
        MessageManager(const string& basedir) 
            :_basedir(basedir)
        {}
        // 初始化指定队列的消息管理类句柄
        void initQueueMessage(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                // 首先需要找一下，如果句柄已经存在，就没有再次创建的需要了
                auto it = _queue_msgs.find(qname);

                if (it != _queue_msgs.end()) {
                    return;
                }
                // 构造对象

                qmg = make_shared<QueueMessage>(_basedir, qname);

                // 新增
                _queue_msgs.insert(make_pair(qname, qmg));
            }
            // 恢复历史数据
            qmg->recovery();
        }

        // 销毁指定队列的消息管理类句柄
        void destroyQueueMessage(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                // 也是需要先找一下，如果已经不存在了，就没有必要再次删除了
                if (it == _queue_msgs.end()) {
                    return ;
                }
                qmg = it->second;
                
                _queue_msgs.erase(qname);
            }
            // 删除句柄数据
            qmg->clear();
        }

        // 向指定队列新增一条消息
        bool insert(const string& qname, const mqNs::BasicProperties* bp, const string& body, bool queue_is_durable) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "行队列: " << qname << " 新增消息失败: 没有找到对应的管理句柄!" << endl;
                    return false;
                }
                qmg = it->second;
            }
            qmg->insert(bp, body, queue_is_durable);
            return true;
        }

        // 获取队首消息
        MessagePtr front(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                MessagePtr msg;
                if (_queue_msgs.empty()) {
                    LOG(Waring) << "队列为空, 消息为空不能获取队首消息!" << endl;
                    return msg;
                }
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "获取队列: " << qname << " 队首消息失败: 没有找到对应的管理句柄!" << endl;
                    return msg;
                }
                qmg = it->second;
            }

            return qmg->front();
        }

        // 对指定队列里的指定的一条消息进行确认
        void ack(const string& qname, const string& msg_id) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "对队列: " << qname << " 指定消息: " << msg_id << " 进行确认失败: 没有找到对应的管理句柄!" << endl;
                    return;
                }
                qmg = it->second;
            }

            qmg->remove(msg_id);
        }


        // 获取可推送消息数量
        size_t push_count(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "获取队列: " << qname << " 可推送消息数量失败: 没有找到对应的管理句柄!" << endl;
                    return 0;
                }
                qmg = it->second;
            }

            return qmg->push_count();
        }

        // 获取持久化存储的消息数量
        size_t durable_count(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "获取队列: " << qname << " 持久化存储的消息数量失败: 没有找到对应的管理句柄!" << endl;
                    return 0;
                }
                qmg = it->second;
            }

            return qmg->durable_count();
        }
        
        // 获取总体的消息数量
        size_t total_count(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "获取队列: " << qname << " 总体的消息数量失败: 没有找到对应的管理句柄!" << endl;
                    return 0;
                }
                qmg = it->second;
            }

            return qmg->total_count();
        }
        // 获取待确认的消息数量
        size_t waitack_count(const string& qname) {
            QueueMessage::ptr qmg;
            {
                unique_lock<mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    LOG(Debug) << "获取队列: " << qname << " 待确认的消息数量失败: 没有找到对应的管理句柄!" << endl;
                    return 0;
                }
                qmg = it->second;
            }
            return qmg->waitack_count();    
        }
        void clear() {
            unique_lock<mutex> lock(_mutex);
            for (auto& msgp : _queue_msgs) {
                msgp.second->clear();
            }
        }
    private :
        mutex _mutex;
        string _basedir; // 所有消息文件存储的目录
        unordered_map<string, QueueMessage::ptr> _queue_msgs; // 队列名称与其消息管理类句柄的映射哈希表
    };
};